package net.bwie.realtime.warehouse.DWS;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.warehouse.DWS
 * @Author: zhangleqing
 * @CreateTime: 2025-09-02  10:36
 * @Description: TODO  基于dwd_before_retreating(退前预警汇总表) 做一个特征衍生
 * 把 需要统计的指标先映射成特征，方便后面的大量计算
 * @Version: 1.0
 */
public class DwsOne {
    private static TableEnvironment tableEnv;

    public static void main(String[] args) {
        // 1.开启上下文
        TableEnvironment tableEnv = getTableEnv();

        // 2.数据读取
        readTable(tableEnv);

        // 3.数据处理
        Table resultTable = handle(tableEnv);

//        resultTable.execute().print();

        // 4.映射表创建
        createView(tableEnv);

        // 5.数据写出
        savaToSink(tableEnv, resultTable);
    }

    private static void savaToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("DwsOne", resultTable);
        tableEnv.executeSql(
                "insert into dws_before_retreating " +
                        "select * from DwsOne"
        ).print();
    }

/*
create table dws_before_retreating(
    order_id STRING,
    product_id STRING,
    pay_time TIMESTAMP(3),
    promise_ship_time TIMESTAMP(3),

    refund_id STRING,
    sender_area_id STRING,
    receiver_area_id STRING,
    logistics_start_time TIMESTAMP(3),
    logistics_end_time TIMESTAMP(3),
    expected_start_time TIMESTAMP(3),
    expected_end_time TIMESTAMP(3),
    logistics_node STRING,
    is_about_to_timeout INT,
    is_late_shipment INT,
    is_stockout INT,
    is_fake_ship_click INT,
    is_collection_update_exception INT,
    is_transport_delivery_exception INT,
    is_delivery_sign_exception INT
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'dws_before_retreating',
    'properties.bootstrap.servers' = 'node101:9092',
    'key.format' = 'json',
    'value.format' = 'json'
)
 */

    private static void createView(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "create table dws_before_retreating(\n" +
                        "    order_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "\n" +
                        "    logistics_start_time TIMESTAMP(3),\n" +
                        "    logistics_end_time TIMESTAMP(3),\n" +
                        "    expected_start_time TIMESTAMP(3),\n" +
                        "    expected_end_time TIMESTAMP(3),\n" +
                        "    logistics_node STRING,\n" +
                        "    is_about_to_timeout INT,\n" +
                        "    is_late_shipment INT,\n" +
                        "    is_stockout INT,\n" +
                        "    is_fake_ship_click INT,\n" +
                        "    is_collection_update_exception INT,\n" +
                        "    is_transport_delivery_exception INT,\n" +
                        "    is_delivery_sign_exception INT,\n" +
                        "    PRIMARY KEY (order_id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dws_before_retreating',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );
    }


/*
SELECT
    TUMBLE_START(logistics_start_time, INTERVAL '1' DAY) AS window_start,
    TUMBLE_END(logistics_start_time, INTERVAL '1' DAY) AS window_end,
    product_id,
    COUNT(DISTINCT order_id) AS order_count
FROM dwd_before_retreating
GROUP BY
    TUMBLE(logistics_start_time, INTERVAL '1' DAY),
    product_id

select
    order_id,
    product_id,
    pay_time,
    promise_ship_time,
    refund_id,
    sender_area_id,
    receiver_area_id,
    logistics_start_time,
    logistics_end_time,
    expected_start_time,
    expected_end_time,
    logistics_node,
    area_id,
    sender_same_province_timeout,
    sender_cross_province_timeout,
    sender_special_area_flag,
    case when
from dwd_before_retreating

-- REGEXP_INSTR 不支持 版本问题 解决办法 1使用自定义函数 2升级版本 3其他解决办法
-- 步骤1：先从logistics_node提取“派送时间”，用CTE复用逻辑（避免重复代码）
WITH node_time_extract AS (
    SELECT
        order_id,
        product_id,
        pay_time,
        promise_ship_time,
        refund_id,
        sender_area_id,
        receiver_area_id,
        logistics_start_time,
        logistics_end_time,
        expected_start_time,
        expected_end_time,
        logistics_node,
        area_id,
        sender_same_province_timeout,
        sender_cross_province_timeout,
        sender_special_area_flag,
        -- 使用自定义函数替代 REGEXP_INSTR
        SUBSTRING(
            logistics_node,
            REGEXP_INSTR(logistics_node, '\\\\d{4}-\\\\d{2}-\\\\d{2} \\\\d{2}:\\\\d{2}:\\\\d{2}'),
            19
        ) AS delivery_time_str,
        CASE WHEN LOCATE('派送', logistics_node) > 0 THEN 1 ELSE 0 END AS is_delivery_node
    FROM dwd_before_retreating
)
-- 步骤2：衍生所有7个特征（主查询）
SELECT
    order_id,
    product_id,
    pay_time,
    promise_ship_time,
    logistics_start_time,
    logistics_end_time,
    expected_start_time,
    expected_end_time,
    logistics_node,
    -- 1. 即将超时：支付时间 - 承诺发货时间 < 8小时
    CASE
        WHEN promise_ship_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, promise_ship_time, pay_time) < 8
        THEN 1
        ELSE 0
    END AS is_about_to_timeout,
    -- 2. 延迟发货：= 揽收时间 - 延迟发货时间 > 8小时
    CASE
        WHEN logistics_start_time IS NOT NULL
             AND promise_ship_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, promise_ship_time, logistics_start_time) > 8
        THEN 1
        ELSE 0
    END AS is_late_shipment,
    -- 3. 缺货：= 揽收时间 - 延迟发货时间 > 72
    CASE
        WHEN logistics_start_time IS NOT NULL
             AND promise_ship_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, promise_ship_time, logistics_start_time) > 72
        THEN 1
        ELSE 0
    END AS is_stockout,
    -- 4. 虚假点击发货：= 揽收时间 - 已发货时间 > 24
    CASE
        WHEN logistics_start_time IS NOT NULL
             AND expected_start_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, expected_start_time, logistics_start_time) > 24
        THEN 1
        ELSE 0
    END AS is_fake_ship_click,
    -- 5. 揽收-更新异常：= 运输 - 揽收 >  24h
    CASE
        WHEN expected_end_time IS NOT NULL
             AND logistics_start_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, logistics_start_time, expected_end_time) > 24
        THEN 1
        ELSE 0
    END AS is_collection_update_exception,
    -- 6. 运输-派送异常：= 派送 - 运输 >  8小时
    CASE
        WHEN is_delivery_node = 1
             AND delivery_time_str IS NOT NULL
             AND TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss') IS NOT NULL
             AND expected_end_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, expected_end_time, TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss')) > 8
        THEN 1
        ELSE 0
    END AS is_transport_delivery_exception,
    -- 7. 派送-签收异常：= 签收时间 - 派送时间 >  24小时
    CASE
        WHEN is_delivery_node = 1
             AND delivery_time_str IS NOT NULL
             AND TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss') IS NOT NULL
             AND logistics_end_time IS NOT NULL
             AND TIMESTAMPDIFF(HOUR, TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss'), logistics_end_time) > 24
        THEN 1
        ELSE 0
    END AS is_delivery_sign_exception
FROM node_time_extract


 */

    private static Table handle(TableEnvironment tableEnv) {
        // 注册自定义函数
        tableEnv.createTemporarySystemFunction("REGEXP_INSTR", new RegExpInstrFunction());

        Table table = tableEnv.sqlQuery(
                "WITH node_time_extract AS (\n" +
                        "    SELECT\n" +
                        "        order_id,\n" +
                        "        product_id,\n" +
                        "        pay_time,\n" +
                        "        promise_ship_time,\n" +
                        "        refund_id,\n" +
                        "        sender_area_id,\n" +
                        "        receiver_area_id,\n" +
                        "        logistics_start_time,\n" +
                        "        logistics_end_time,\n" +
                        "        expected_start_time,\n" +
                        "        expected_end_time,\n" +
                        "        logistics_node,\n" +
                        "        area_id,\n" +
                        "        sender_same_province_timeout,\n" +
                        "        sender_cross_province_timeout,\n" +
                        "        sender_special_area_flag,\n" +
                        "        -- 使用自定义函数替代 REGEXP_INSTR\n" +
                        "        SUBSTRING(\n" +
                        "            logistics_node,\n" +
                        "            REGEXP_INSTR(logistics_node, '\\\\\\\\d{4}-\\\\\\\\d{2}-\\\\\\\\d{2} \\\\\\\\d{2}:\\\\\\\\d{2}:\\\\\\\\d{2}'),\n" +
                        "            19\n" +
                        "        ) AS delivery_time_str,\n" +
                        "        CASE WHEN LOCATE('派送', logistics_node) > 0 THEN 1 ELSE 0 END AS is_delivery_node\n" +
                        "    FROM dwd_before_retreating\n" +
                        ")\n" +
                        "-- 步骤2：衍生所有7个特征（主查询）\n" +
                        "SELECT\n" +
                        "    order_id,\n" +
                        "    product_id,\n" +
                        "    pay_time,\n" +
                        "    promise_ship_time,\n" +
                        "    logistics_start_time,\n" +
                        "    logistics_end_time,\n" +
                        "    expected_start_time,\n" +
                        "    expected_end_time,\n" +
                        "    logistics_node,\n" +
                        "    -- 1. 即将超时：支付时间 - 承诺发货时间 < 8小时\n" +
                        "    CASE\n" +
                        "        WHEN promise_ship_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, promise_ship_time, pay_time) < 8 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_about_to_timeout,\n" +
                        "    -- 2. 延迟发货：= 揽收时间 - 延迟发货时间 > 8小时\n" +
                        "    CASE\n" +
                        "        WHEN logistics_start_time IS NOT NULL \n" +
                        "             AND promise_ship_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, promise_ship_time, logistics_start_time) > 8 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_late_shipment,\n" +
                        "    -- 3. 缺货：= 揽收时间 - 延迟发货时间 > 72\n" +
                        "    CASE\n" +
                        "        WHEN logistics_start_time IS NOT NULL \n" +
                        "             AND promise_ship_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, promise_ship_time, logistics_start_time) > 72 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_stockout,\n" +
                        "    -- 4. 虚假点击发货：= 揽收时间 - 已发货时间 > 24\n" +
                        "    CASE\n" +
                        "        WHEN logistics_start_time IS NOT NULL \n" +
                        "             AND expected_start_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, expected_start_time, logistics_start_time) > 24 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_fake_ship_click,\n" +
                        "    -- 5. 揽收-更新异常：= 运输 - 揽收 >  24h\n" +
                        "    CASE\n" +
                        "        WHEN expected_end_time IS NOT NULL \n" +
                        "             AND logistics_start_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, logistics_start_time, expected_end_time) > 24 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_collection_update_exception,\n" +
                        "    -- 6. 运输-派送异常：= 派送 - 运输 >  8小时\n" +
                        "    CASE\n" +
                        "        WHEN is_delivery_node = 1 \n" +
                        "             AND delivery_time_str IS NOT NULL \n" +
                        "             AND TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss') IS NOT NULL \n" +
                        "             AND expected_end_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, expected_end_time, TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss')) > 8 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_transport_delivery_exception,\n" +
                        "    -- 7. 派送-签收异常：= 签收时间 - 派送时间 >  24小时\n" +
                        "    CASE\n" +
                        "        WHEN is_delivery_node = 1 \n" +
                        "             AND delivery_time_str IS NOT NULL \n" +
                        "             AND TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss') IS NOT NULL \n" +
                        "             AND logistics_end_time IS NOT NULL \n" +
                        "             AND TIMESTAMPDIFF(HOUR, TO_TIMESTAMP(delivery_time_str, 'yyyy-MM-dd HH:mm:ss'), logistics_end_time) > 24 \n" +
                        "        THEN 1 \n" +
                        "        ELSE 0 \n" +
                        "    END AS is_delivery_sign_exception\n" +
                        "FROM node_time_extract"
        );
        // 4. 虚假点击发货
        // 6. 运输-派送异常
        // 7. 派送-签收异常
        return table;
    }

    private static void readTable(TableEnvironment tableEnv) {
        tableEnv.executeSql(
                "CREATE TABLE dwd_before_retreating (\n" +
                        "    order_id STRING,\n" +
                        "    product_id STRING,\n" +
                        "    pay_time TIMESTAMP(3),\n" +
                        "    promise_ship_time TIMESTAMP(3),\n" +
                        "    refund_id STRING,\n" +
                        "    sender_area_id STRING,\n" +
                        "    receiver_area_id STRING,\n" +
                        "    logistics_start_time TIMESTAMP(3),  -- 事件时间字段\n" +
                        "    logistics_end_time TIMESTAMP(3),\n" +
                        "    expected_start_time TIMESTAMP(3),\n" +
                        "    expected_end_time TIMESTAMP(3),\n" +
                        "    logistics_node STRING,\n" +
                        "    area_id STRING,\n" +
                        "    sender_same_province_timeout INT,\n" +
                        "    sender_cross_province_timeout INT,\n" +
                        "    sender_special_area_flag INT,\n" +
                        "    -- 定义水位线：基于logistics_start_time，允许10分钟延迟（可根据业务调整）\n" +
                        "    WATERMARK FOR logistics_start_time AS logistics_start_time - INTERVAL '10' MINUTE,\n" +
                        "    PRIMARY KEY (order_id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dwd_before_retreating',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json',\n" +
                        "    -- 如果需要从最早位置消费，可添加以下配置\n" +
                        "    'properties.auto.offset.reset' = 'earliest'\n" +
                        ")"
        );
    }


    public static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态TTL设置为25小时（3600*25=90000秒），覆盖24小时的关联区间
        configuration.setString("table.exec.state.ttl", "90000 s");
//        configuration.setString("execution.checkpointing.interval", "5 s");
        // 3.返回对象
        return tabEnv;
    }
}
